# --------------------------------------------------------------------------------------------------------------------
# Project:             Digital Payment System (DPS)
# Website:             http://code.google.com/p/dps-x509/
# Purpose of document: Web service test tool for registrar
# --------------------------------------------------------------------------------------------------------------------
import urllib, sys

try:
  import simplejson as json
except ImportError:
  import json
  
HOST = ""

# --------------------------------------------------------------------------------------------------------------------
def callWebService(name, params, exp):
  """ call the web service and compare expected and actual results
      @param name   - name of the web service to call
      @param params - struct of input parameters
      @param exp    - json str formatted expected results
      @return success, message tuple. success = true if expected result = actual result """
  #print ("executing "+name+" ...");
  res = False
  message = ""
  try:
    global HOST
    #print len(params)
    p = urllib.urlencode(params)
    #print params    
    f = urllib.urlopen(HOST + name, p)
    actual = urllib.unquote(f.read())
    f.close()
    if actual == exp:
      res = True
    else:
      message = "failed. actual len: "+str(len(actual))+" expected:"+str(len(exp))+"\n"
      message += "actual:\n"+actual+"\nexpected:\n"+exp
  except:
    message = "Error connecting to web service"
  return res, message    
  
# --------------------------------------------------------------------------------------------------------------------
class TestSuite:
  """ container holding all tests """
  def __init__(self):
    self.tests_ = []
  
  def addTest(self, t):
    """ add a test to the suite. 
    @param t Test must be a bool,string returning function that takes no params """
    self.tests_.append(t)
    
  def runTests(self):
    """ run all tests in suite """
    passed = 0    
    for t in self.tests_:
      msg = ""      
      didPass = False
      resText = "False"
      try:
        didPass, msg = t()
        if didPass:
          passed += 1
          resText = "True"
      except Exception, e:
        msg = "An exception occured: " + str(sys.exc_info()[0]) + " " + str(e)
      print("\n*   %s: %s" % (t.__name__, resText))
      if not didPass:
        print(msg)
    print("\n*** Tests passed: %d of %d ***" % (passed, len(self.tests_)))

# --------------------------------------------------------------------------------------------------------------------
# --------------------------------------------------------------------------------------------------------------------
def clearCertificate():
  result = json.dumps({"returnCode": "True"})  
  isMatch, msg = callWebService("registrar/clear", {"format": "json"}, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64_2048_lengthPublicKeyGood():
  cert = open("testbase64.crt", "rb").read()
  #print cert
  params = {"certificate":cert, "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def certificateAlreadyInRegistrar():
  isOk, msg = clearCertificate()
  if not isOk:
    return False, "Could not clear registry"
  
  isOk, msg = base64_2048_lengthPublicKeyGood()
  if not isOk:
    return False, "Could not add first certificate"
  
  cert = open("testbase64.crt", "rb").read()
  params = {"certificate":cert, "format": "json"}
  result = json.dumps({"returnCode": "Rejected:CertificateAlreadyInRegistrarError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64_1024_lengthPublicKeyGood():
  cert = open("testbase64_1024pubkey.crt", "rb").read()
  #print cert
  params = {"certificate":cert, "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64dsa():
  cert = open("testbase64dsa.crt", "rb").read()
  
  params = {"certificate":cert, "format": "json"}
  result = json.dumps({"returnCode": "Rejected:NoRSADataFoundError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64BadSignature():
  cert = open("testbase64.crt", "rb").read()
  #switch two values
  pos = 44
  val = cert[pos:pos+1]
  if val == "a":
    val = "b"
  else:
    val = "a"
  newCert = cert[:pos] + val + cert[pos+1:]
  
  params = {"certificate":newCert, "format": "json"}
  result = json.dumps({"returnCode": "Rejected:InvalidSignatureError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64ExtraCharacters():
  cert = open("testbase64.crt", "rb").read()
  #print cert
  #print len(cert)
  pos = 40
  newCert = cert[:pos] + "extra" + cert[pos:]
  params = {"certificate":newCert, "format": "json"}
  result = json.dumps({"returnCode": "Rejected:DecodingError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def derGood():
  cert = open("testder.crt", "rb").read()
  #print cert
  #print len(cert)
  params = {"certificate":cert, "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg
  
# --------------------------------------------------------------------------------------------------------------------
def nonSelfSigned():
  cert = open("../../../../../Certificates/SecondRecipientSingleCertificate.crt", "rb").read()
  #print cert
  #print len(cert)
  params = {"certificate":cert, "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------

# --------------------------------------------------------------------------------------------------------------------
def runTests(argv):
  global HOST
  #HOST = "http://dps-x509.appspot.com/"
  HOST = "http://localhost:8083/"
  
  if len(argv) > 1:
    HOST = argv[1]
  print("host: "+HOST)  
  
  ts = TestSuite()

  ts.addTest(clearCertificate)
  ts.addTest(base64_2048_lengthPublicKeyGood)
  ts.addTest(base64_1024_lengthPublicKeyGood)
  ts.addTest(certificateAlreadyInRegistrar)
  ts.addTest(base64dsa)
  ts.addTest(base64BadSignature)
  ts.addTest(base64ExtraCharacters)
  ts.addTest(derGood) 
  ts.addTest(nonSelfSigned)

  ts.runTests()
  
# --------------------------------------------------------------------------------------------------------------------
if __name__ == "__main__":
  runTests(sys.argv)

